package com.gzhryc.shared_device.oem.micro.queue.work;

import java.util.Date;
import java.util.List;

import org.apache.commons.lang3.time.DateUtils;

import com.gzhryc.common.json.JsonTools;
import com.gzhryc.common.kafka.JFKafkaProducer;
import com.gzhryc.common.kafka.KafkaProducerFactory;
import com.gzhryc.common.kafka.models.KafkaMessage;
import com.gzhryc.common.logging.Logger;
import com.gzhryc.shared_device.base.models.OperateInfo;
import com.gzhryc.shared_device.oem.code.KafkaEventConstants;
import com.gzhryc.shared_device.oem.code.PayServiceFactory;
import com.gzhryc.shared_device.oem.code.business.EBusinessType;
import com.gzhryc.shared_device.oem.code.member.services.MemberBusinessOrderService;
import com.gzhryc.shared_device.oem.code.order.RedisCacheManager;
import com.gzhryc.shared_device.oem.code.order.event.OrderFinishEvent;
import com.gzhryc.shared_device.pcl.code.dao.db.PCLOrder;
import com.gzhryc.shared_device.pcl.code.dao.enums.PCLOrderEnum;
import com.gzhryc.shared_device.pcl.code.services.PCLOrderService;
import com.gzhryc.shared_device.pcl.code.services.dto.PCLOrderSearch;

public class PCLFinishCheckRunnable implements Runnable{

    static Logger log = Logger.getLogger(PCLFinishCheckRunnable.class);

    OperateInfo operateInfo;

    public PCLFinishCheckRunnable(OperateInfo operateInfo){
        this.operateInfo = operateInfo;
    }

	@Override
	public void run() {
		Date currentDate = DateUtils.addMinutes(new Date(),30);		//30分钟误差
		
		Integer pageNo = 1;
		Integer pageLimit = 100;
		
		PCLOrderSearch search = new PCLOrderSearch();
		search.setState(PCLOrderEnum.State.Working.index());
		search.setMaxEndDate(currentDate);
		
		//时间已到
        operateInfo.setOperateNote("便携线充订单服务时间已到");
		
		List<PCLOrder> pclOrderList = PCLOrderService.self(currentDate).findBySearch(search, pageNo, pageLimit);
		while(pclOrderList != null && pclOrderList.size() > 0) {
			for(PCLOrder deviceOrder : pclOrderList) {
				try {
                    int result = PCLOrderService.self(deviceOrder.getCreateDate()).finish(deviceOrder, deviceOrder.getPayMoney(), PayServiceFactory.self(), operateInfo);
                    if (result == 2) {
                        //订单已经完成，同步会员订单
                        deviceOrder = PCLOrderService.self(deviceOrder.getCreateDate()).getByTradeNo(deviceOrder.getTradeNo());
                        if (MemberBusinessOrderService.self(deviceOrder.getCreateDate()).finish(deviceOrder.getTradeNo(), deviceOrder.getMoney(), deviceOrder.getPayMoney()
                                , deviceOrder.getWalletMoney(),operateInfo)) {
                            log.error("便携线充订单完成，但同步会员订单失败，订单号：{{0}}", deviceOrder.getTradeNo());
                        }

                        JFKafkaProducer producer = KafkaProducerFactory.self().getProducer();
                        if (producer != null) {
                            OrderFinishEvent event = new OrderFinishEvent(EBusinessType.PCL.index(), deviceOrder.getTradeNo(), deviceOrder.getCreateDate(), operateInfo);
                            KafkaMessage kafkaMessage = new KafkaMessage(KafkaEventConstants.TOPIC_Order, KafkaEventConstants.KEY_OrderFinish, JsonTools.toJson(event));
                            producer.send(kafkaMessage);
                        }else{
                            log.error("未初始化消息队列，订单{{0}}未触发完成事件",deviceOrder.getTradeNo());
                        }
                    }
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                } finally {
                	//清理缓存，避免定时器重复处理
                    RedisCacheManager.removeWorkingPCLDeviceOrder(deviceOrder.getTradeNo());
                }
			}
			pageNo = pageNo + 1;
			pclOrderList = PCLOrderService.self(currentDate).findBySearch(search, pageNo, pageLimit);
		}
	}


}
